package com.bw.gmall.realtime.dws.app;

import cn.hutool.core.collection.ListUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.base.BaseApp;
import com.bw.gmall.realtime.common.bean.TrafficHomeDetailPageViewBean;
import com.bw.gmall.realtime.common.bean.TrafficPageViewBean;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.DateFormatUtil;
import com.bw.gmall.realtime.common.util.DorisMapFunction;
import com.bw.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;

/**
 * @projectName: gmalll2206
 * @package: com.bw.gmall.realtime.dws.app
 * @className: DwsTrafficHomeDetailPageViewWindow
 * @author: DuanYu
 * @description: TODO
 * @date: 2025/3/3 9:45
 * @version: 1.0
 */

public class DwsTrafficHomeDetailPageViewWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsTrafficHomeDetailPageViewWindow().start(Constant.TOPIC_DWD_TRAFFIC_PAGE, Constant.DWS_TRAFFIC_HOME_DETAIL_PAGE_VIEW_WINDOW, 4, 10023);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {
        SingleOutputStreamOperator<JSONObject> getEtl = getGetEtl(dataStreamSource);
//        getEtl.print();
        SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> keyByStream = keyByStream(getEtl);
        //        keyByStream.print();
        SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> wmStream = getWmStream(keyByStream);
        SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> process = getProcess(wmStream);
//        process.print();
        process.map(new DorisMapFunction<>()).sinkTo(FlinkSinkUtil.getDorisSink(Constant.DWS_TRAFFIC_HOME_DETAIL_PAGE_VIEW_WINDOW));

    }

    private static SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> getWmStream(SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> keyByStream) {
        return keyByStream.assignTimestampsAndWatermarks(WatermarkStrategy.<TrafficHomeDetailPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(5)).withTimestampAssigner(new SerializableTimestampAssigner<TrafficHomeDetailPageViewBean>() {
            @Override
            public long extractTimestamp(TrafficHomeDetailPageViewBean trafficHomeDetailPageViewBean, long l) {
                return trafficHomeDetailPageViewBean.getTs();
            }
        }));
    }

    private static SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> getProcess(SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> keyByStream) {
        return keyByStream.windowAll(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(5))).process(new ProcessAllWindowFunction<TrafficHomeDetailPageViewBean, TrafficHomeDetailPageViewBean, TimeWindow>() {
            @Override
            public void process(ProcessAllWindowFunction<TrafficHomeDetailPageViewBean, TrafficHomeDetailPageViewBean, TimeWindow>.Context context, Iterable<TrafficHomeDetailPageViewBean> iterable, Collector<TrafficHomeDetailPageViewBean> collector) throws Exception {
                ArrayList<TrafficHomeDetailPageViewBean> list = ListUtil.toList(iterable);
                long homeCt = list.stream().mapToLong(e -> e.getHomeUvCt()).sum();
                long gd = list.stream().mapToLong(e -> e.getGoodDetailUvCt()).sum();
                long start = context.window().getStart();
                String sst = DateFormatUtil.tsToDateTime(start);
                long end = context.window().getEnd();
                String edt = DateFormatUtil.tsToDateTime(end);
                String cur_data = DateFormatUtil.tsToDate(System.currentTimeMillis());
                TrafficHomeDetailPageViewBean next = list.iterator().next();
                TrafficHomeDetailPageViewBean build = TrafficHomeDetailPageViewBean.builder().stt(sst).edt(edt).homeUvCt(homeCt)
                        .goodDetailUvCt(gd).curDate(cur_data).ts(next.getTs()).build();
                collector.collect(build);
            }
        });
    }

    private static SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> keyByStream(SingleOutputStreamOperator<JSONObject> getEtl) {
        return getEtl.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject jsonObject) throws Exception {
                return jsonObject.getJSONObject("common").getString("mid");
            }
        }).process(new KeyedProcessFunction<String, JSONObject, TrafficHomeDetailPageViewBean>() {
            private ValueState<String> homeState;
            private ValueState<String> detailState;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> homeStateDesc = new ValueStateDescriptor<>("homeVale", String.class);
                homeStateDesc.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite).build());
                homeState = getRuntimeContext().getState(homeStateDesc);
                ValueStateDescriptor<String> detailStateDesc = new ValueStateDescriptor<>("detailState", String.class);
                detailStateDesc.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite).build());
                detailState = getRuntimeContext().getState(detailStateDesc);
            }

            @Override
            public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, TrafficHomeDetailPageViewBean>.Context context, Collector<TrafficHomeDetailPageViewBean> collector) throws Exception {
                //取出ts时间戳
                Long ts = jsonObject.getLong("ts");
                //转换为日期类型
                String s = DateFormatUtil.tsToDate(ts);
                //取出page_id
                JSONObject page = jsonObject.getJSONObject("page");
                String page_id = page.getString("page_id");
                //获取状态的值
                String homeValue = homeState.value();
                String detailValue = detailState.value();
                //初始化homeCt 和 detailCt
                Long homeCt = 0L;
                Long datailCt = 0L;
                if ("home".equals(page_id)) {
                    if (!s.equals(homeValue)) {
                        homeCt = 1L;
                        homeState.update(s);
                    }
                } else if ("good_detail".equals(page_id)) {
                    if (!s.equals(detailValue)) {
                        datailCt = 1L;
                        detailState.update(s);
                    }
                }
                if (homeCt != 0 || datailCt != 0) {
                    TrafficHomeDetailPageViewBean build = TrafficHomeDetailPageViewBean.builder().homeUvCt(homeCt).goodDetailUvCt(datailCt).ts(ts).build();
                    collector.collect(build);
                }
            }
        });
    }

    private static SingleOutputStreamOperator<JSONObject> getGetEtl(DataStreamSource<String> dataStreamSource) {
        return dataStreamSource.flatMap(new RichFlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
                if (s != null) {
                    JSONObject jsonObject = JSONObject.parseObject(s);
                    JSONObject common = jsonObject.getJSONObject("common");

                    JSONObject page = jsonObject.getJSONObject("page");
                    String ts = jsonObject.getString("ts");
                    //过滤出如果page不为空 并且 mid不为空的数据 ts不为空的数据
                    if (common != null && page!=null && ts!=null){
                        String mid = common.getString("mid");
                        String page_id = page.getString("page_id");
                        if (mid != null && page_id!=null){
                            if ("home".equals(page_id) || "good_detail".equals(page_id)){
                                collector.collect(jsonObject);
                            }
                        }
                    }
                }
            }
        });
    }
}
